"""
Case Type   : 发布订阅
Case Name   : 发布订阅过程中磁盘满
Create At   : 2022-11-10
Owner       : peilinqian
Description :
    1.两个集群创建表
    2.创建发布订阅
    3.发布端注入磁盘满故障
    4.磁盘满过程中发布端更新数据
    5.订阅端查询数据是否更新
    6.清理磁盘满故障,并重启数据库
    7.发布端更新数据
    8.订阅端查询数据是否更新
    9.订阅端注入磁盘满故障
    10.磁盘满过程中发布端更新数据
    11.订阅端查询数据是否更新
    12.清理磁盘满故障,并重启数据库
    13.订阅端查询数据是否更新
    14.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.合理报错
    5.不更新
    6.成功
    7.成功
    8.更新
    9.成功
    10.成功
    11.订阅端集群异常
    12.成功
    13.更新
    14.成功
History     : 
"""

import os
import unittest
from yat.test import macro
from yat.test import Node
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("-----------this is setup-----------")
        self.log.info(f"-----{os.path.basename(__file__)} start-----")
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_userdb_pub_root = Node(node='PrimaryRoot')
        self.pri_userdb_sub = Node(node='remote_PrimaryDbUser')
        self.pri_userdb_sub_root = Node(node='remote_PrimaryRoot')
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_sub = CommonSH('remote_PrimaryDbUser')
        self.tb_name = 't_pub_sub_reliability_0003'
        self.subname = "s_pub_sub_reliability_0003"
        self.pubname = "p_pub_sub_reliability_0003"
        self.parent_path_pub = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.parent_path_sub = os.path.dirname(macro.DB_INSTANCE_PATH_REMOTE)
        self.port = str(int(self.pri_userdb_pub.db_port) + 1)
        result = self.commsh_pub.executDbSql("show wal_level;")
        self.wal_level = str(result.splitlines()[-2]).strip()
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
            f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
            f'-W {self.pri_userdb_sub.db_password}'
        cmd = f"cp " \
            f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')} " \
            f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        cmd = f"cp " \
            f"{os.path.join(macro.DB_INSTANCE_PATH_REMOTE, 'pg_hba.conf')}" \
            f" {os.path.join(self.parent_path_sub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        self.hostname = self.pri_userdb_sub.sh('hostname').result().strip()
        text = "--step:预置条件,修改pg_hba expect:成功"
        self.log.info(text)
        guc_res = self.commsh_pub.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        guc_res = self.commsh_pub.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    all  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        result = self.commsh_pub.executeGsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
        self.assertTrue(result, '执行失败' + text)
        pub_result = self.commsh_pub.stopDbCluster(get_detail=True)
        pub_result1 = self.commsh_pub.startDbCluster(get_detail=True)
        self.assertTrue(pub_result, '执行失败' + text)
        flg = self.constant.START_SUCCESS_MSG in pub_result1 or 'Degrade' in \
              pub_result1
        self.assertTrue(flg, '执行失败' + text)
        guc_res = self.commsh_sub.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        guc_res = self.commsh_sub.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE,
            f'host all  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)

    def test_pub_sub(self):
        text = "--step1:在两个集群创建表 expect:成功--"
        self.log.info(text)
        sql_cmd = f"drop table if exists {self.tb_name};" \
            f"create table {self.tb_name}(c_int int primary key,c_text text);"
        pub_result = self.commsh_pub.executDbSql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertEqual(pub_result.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败' + text)
        sub_result = self.commsh_sub.executDbSql(
            sql_cmd, self.user_param_sub, None,
            env_path=macro.DB_ENV_PATH_REMOTE)
        self.log.info(sub_result)
        self.assertEqual(sub_result.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败' + text)

        text = "--step2:创建发布订阅 expect:成功--"
        self.log.info(text)
        sql_cmd = f"drop publication if exists {self.pubname};" \
            f"create publication {self.pubname} for all tables;"
        pub_result = self.commsh_pub.executDbSql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertIn(self.constant.create_pub_succ_msg, pub_result,
                      '执行失败' + text)
        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE)
        self.assertIn(self.constant.create_keycipher_success, result,
                      '执行失败' + text)
        sql_cmd = f"drop subscription if exists {self.subname};" \
            f"create subscription {self.subname} connection " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.pri_userdb_pub.db_name} " \
            f"password={self.pri_userdb_pub.ssh_password}' publication " \
            f"{self.pubname};"
        sub_result = self.commsh_sub.executDbSql(
            sql_cmd, self.user_param_sub, None,
            env_path=macro.DB_ENV_PATH_REMOTE)
        self.log.info(sub_result)
        self.assertIn(self.constant.create_sub_succ_msg, sub_result, '执行失败'
                      + text)

        try:
            text = "--step3:发布端注入磁盘满故障 expect:成功--"
            self.log.info(text)
            cfe_inject_cmd = f"{macro.CFE_PATH}/cfe 'inject rfile_full " \
                f"(diskname) values ({self.dir_name})'"
            self.log.info(cfe_inject_cmd)
            cfe_inject_res = self.pri_userdb_pub_root.\
                sh(cfe_inject_cmd).result()
            self.log.info(cfe_inject_res)
            self.assertIn(self.constant.CFE_DISK_FULL_SUCCESS_MSG,
                          cfe_inject_res, '执行失败' + text)
            df_cmd = f"df -h {self.dir_name}"
            self.log.info(df_cmd)
            df_result = self.pri_userdb_pub_root.sh(df_cmd).result()
            self.log.info(df_result)
            self.USED_PERCENT = df_result.splitlines()[-1].split()[4].strip()
            self.log.info(self.USED_PERCENT)
            self.assertTrue(self.USED_PERCENT == '100%', '执行失败' + text)

            text = "--step4:磁盘满过程中发布端更新数据 expect:合理报错--"
            self.log.info(text)
            sql_cmd = f"insert into {self.tb_name}(c_int) values" \
                f"(generate_series(1,10));"
            pub_result = self.commsh_pub.executDbSql(
                sql_cmd, sql_type=self.user_param_pub)
            self.log.info(pub_result)
            self.assertIn(self.constant.NO_SPACE_BUILD_FAIL_MSG, pub_result,
                          '执行失败' + text)

            text = "--step5:订阅端查询数据是否更新 expect:不更新--"
            self.log.info(text)
            sql_select = f"select pg_sleep(5.5);select count(*) from " \
                f"{self.tb_name};"
            self.log.info(sql_select)
            sub_result = self.commsh_sub.executDbSql(
                sql_select, self.user_param_sub, False,
                env_path=macro.DB_ENV_PATH_REMOTE)
            self.assertEqual('0', str(sub_result.splitlines()[-2]).strip(),
                             '执行失败' + text)

        finally:
            text = "--step6:清理磁盘满故障,并重启数据库 expect:成功--"
            self.log.info(text)
            cfe_inject_cmd = f"{macro.CFE_PATH}/cfe 'clean rfile_full " \
                f"where(diskname={self.dir_name})'"
            self.log.info(cfe_inject_cmd)
            cfe_inject_res = self.pri_userdb_pub_root.\
                sh(cfe_inject_cmd).result()
            self.log.info(cfe_inject_res)
            self.assertIn(self.constant.CFE_DISK_CLEAN_SUCCESS_MSG,
                          cfe_inject_res, '执行失败' + text)
            df_cmd = f"df -h {self.dir_name}"
            self.log.info(df_cmd)
            df_result = self.pri_userdb_pub_root.sh(df_cmd).result()
            self.log.info(df_result)
            self.USED_PERCENT = df_result.splitlines()[-1].split()[4].strip()
            self.log.info(self.USED_PERCENT)
            self.assertTrue(self.USED_PERCENT != '100%', '执行失败' + text)
            pub_result = self.commsh_pub.stopDbCluster(get_detail=True)
            pub_result1 = self.commsh_pub.startDbCluster(get_detail=True)
            self.assertTrue(pub_result, '执行失败' + text)
            self.assertTrue(pub_result1, '执行失败' + text)

        text = "--step7:发布端更新数据 expect:成功--"
        self.log.info(text)
        sql_cmd = f"insert into {self.tb_name}(c_int) values" \
            f"(generate_series(200,1000));"
        pub_result = self.commsh_pub.executDbSql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertIn(self.constant.insert_success_msg, pub_result,
                      '执行失败' + text)

        text = "--step8:订阅端查询数据是否更新 expect:数据更新--"
        self.log.info(text)
        sql_select = f"select pg_sleep(5.5);select count(c_int) from " \
            f"{self.tb_name};"
        self.log.info(sql_select)
        sub_result = self.commsh_sub.executDbSql(
            sql_select, self.user_param_sub, False,
            env_path=macro.DB_ENV_PATH_REMOTE)
        self.assertEqual('801', str(sub_result.splitlines()[-2]).strip(),
                         '执行失败' + text)

        try:
            text = "--step9:订阅端注入磁盘满故障 expect:成功--"
            self.log.info(text)
            cfe_inject_cmd = f"{macro.CFE_PATH}/cfe 'inject rfile_full " \
                f"(diskname) values ({self.dir_name})'"
            self.log.info(cfe_inject_cmd)
            cfe_inject_res = self.pri_userdb_sub_root.\
                sh(cfe_inject_cmd).result()
            self.log.info(cfe_inject_res)
            self.assertIn(self.constant.CFE_DISK_FULL_SUCCESS_MSG,
                          cfe_inject_res, '执行失败' + text)
            df_cmd = f"df -h {self.dir_name}"
            self.log.info(df_cmd)
            df_result = self.pri_userdb_sub_root.sh(df_cmd).result()
            self.log.info(df_result)
            self.USED_PERCENT = df_result.splitlines()[-1].split()[4].strip()
            self.log.info(self.USED_PERCENT)
            self.assertTrue(self.USED_PERCENT == '100%', '执行失败' + text)

            text = "--step10:磁盘满过程中发布端更新数据 expect:成功--"
            self.log.info(text)
            sql_cmd = f"update {self.tb_name} set c_text='update';"
            pub_result = self.commsh_pub.executDbSql(
                sql_cmd, sql_type=self.user_param_pub)
            self.log.info(pub_result)
            self.assertIn(self.constant.update_success_msg, pub_result,
                          '执行失败' + text)

            text = "--step11:订阅端查询数据是否更新 expect:订阅端集群异常--"
            self.log.info(text)
            sql_select = f"select pg_sleep(5.5);select count(*) from " \
                f"{self.tb_name} where c_text='update';"
            self.log.info(sql_select)
            sub_result = self.commsh_sub.executDbSql(
                sql_select, self.user_param_sub, False,
                env_path=macro.DB_ENV_PATH_REMOTE)
            self.assertIn('connection to server was lost', sub_result,
                          '执行失败' + text)

        finally:
            text = "--step12:清理磁盘满故障,并重启数据库 expect:成功--"
            self.log.info(text)
            cfe_inject_cmd = f"{macro.CFE_PATH}/cfe 'clean rfile_full " \
                f"where(diskname={self.dir_name})'"
            self.log.info(cfe_inject_cmd)
            cfe_inject_res = self.pri_userdb_sub_root.\
                sh(cfe_inject_cmd).result()
            self.log.info(cfe_inject_res)
            self.assertIn(self.constant.CFE_DISK_CLEAN_SUCCESS_MSG,
                          cfe_inject_res, '执行失败' + text)
            df_cmd = f"df -h {self.dir_name}"
            self.log.info(df_cmd)
            df_result = self.pri_userdb_sub_root.sh(df_cmd).result()
            self.log.info(df_result)
            self.USED_PERCENT = df_result.splitlines()[-1].split()[4].strip()
            self.log.info(self.USED_PERCENT)
            self.assertTrue(self.USED_PERCENT != '100%', '执行失败' + text)
            sub_result = self.commsh_sub.stopDbCluster(
                get_detail=True, env_path=macro.DB_ENV_PATH_REMOTE)
            sub_result1 = self.commsh_sub.startDbCluster(
                get_detail=True, env_path=macro.DB_ENV_PATH_REMOTE)
            self.assertTrue(sub_result, '执行失败' + text)
            self.assertTrue(sub_result1, '执行失败' + text)

        text = "--step13:订阅端查询数据是否更新 expect:更新--"
        self.log.info(text)
        sql_select = f"select pg_sleep(5.5);select count(*) from " \
            f"{self.tb_name} where c_text='update';"
        self.log.info(sql_select)
        sub_result = self.commsh_sub.executDbSql(
            sql_select, self.user_param_sub, False,
            env_path=macro.DB_ENV_PATH_REMOTE)
        self.assertEqual(str(sub_result.splitlines()[-2]).strip(), '801',
                         '执行失败' + text)

    def tearDown(self):
        text = "--step14:清理环境 expect:成功--"
        self.log.info(text)
        sql_cmd = f"drop subscription if exists {self.subname};"
        drop_sub_result = self.commsh_sub.executDbSql(
            sql_cmd, self.user_param_sub, None,
            env_path=macro.DB_ENV_PATH_REMOTE)
        self.log.info(drop_sub_result)
        sql_cmd = f"drop publication if exists {self.pubname};"
        drop_pub_result = self.commsh_pub.executDbSql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result)
        sql_cmd = f"drop table if exists {self.tb_name};"
        result_sub = \
            self.commsh_sub.executDbSql(sql_cmd, self.user_param_sub, None,
                                        env_path=macro.DB_ENV_PATH_REMOTE)
        self.log.info(result_sub)
        result_pub = self.commsh_pub.executDbSql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(result_pub)
        cmd = f"mv " \
            f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')} "\
            f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')} "
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        cmd = f"mv " \
            f"{os.path.join(self.parent_path_sub, 'pg_hba.conf')} "\
            f"{os.path.join(macro.DB_INSTANCE_PATH_REMOTE, 'pg_hba.conf')} "
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        result = self.commsh_pub.executeGsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        self.log.info(result)
        pub_result = self.commsh_pub.stopDbCluster(get_detail=True)
        pub_result1 = self.commsh_pub.startDbCluster(get_detail=True)
        self.assertTrue(pub_result, '执行失败' + text)
        self.assertTrue(pub_result1, '执行失败' + text)
        self.assertTrue(result, '执行失败' + text)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, result_pub, '执行失败'
                      + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, result_sub, '执行失败'
                      + text)
        self.log.info(f"-----{os.path.basename(__file__)} end-----")
